// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: fengyang

#pragma once

#include "coin2/exchange/base/feed/update_manager.h"
#include "coin2/exchange/base/feed/subscriber.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/base/executor/lifecycle_timestamp.h"
#include "coin2/exchange/base/feed/counter/counter.h"

#include "coin2/exchange/base/feed/replay/book.h"
#include "coin2/exchange/base/feed/replay/kline.h"
#include "coin2/exchange/base/feed/replay/trading_data.h"

namespace coin2::exchange::base::feed::replay {
namespace impl {

using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::base::feed::FeedUpdateManager;
using coin2::exchange::base::feed::IFeedSubscriber;
using coin2::exchange::base::executor::LifecycleTimestamp;
using coin2::exchange::base::feed::counter::FeedCounter;
using coin2::exchange::base::feed_util::Index;

template<typename T>
struct ReplayMessage {
  int64_t timestamp;  // from packet timestamp
  std::string native_symbol;
  T data;
};

template<typename T>
struct ReplayLifecycleMessage {
  LifecycleTimestamp lts;
  ReplayMessage<T> data;
};

class ReplayProcessor {
 public:
  ReplayProcessor(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      FeedCounter* counter)
      : upd_mgr_(
            system_name,
            mea,
            false,  /* enable_diff_book */
            false  /* enable_telemetry */ ),
        feed_counter_(counter) {
  }

  const FeedUpdateManager& upd_mgr() const  { return upd_mgr_; }
  FeedUpdateManager* mutable_upd_mgr() { return &upd_mgr_; }

  virtual void AddProduct(const IProduct& product, IFeedSubscriber* subscriber) {
    if (subscriber_ == nullptr) {
      subscriber_ = subscriber;
    } else {
      CHECK_EQ(subscriber_, subscriber);
    }
    upd_mgr_.AddProductForFeed(product, subscriber);
  }

  virtual void RemoveProduct(const std::vector<const IProduct*>& products) {
    upd_mgr_.RemoveProductForFeed(products);
  }

  template<typename T>
  void onSharedMessage(std::shared_ptr<ReplayMessage<T>> message) {
    onMessage<T>(*message);
  }

  template<typename T>
  inline void onMessage(const ReplayMessage<T>& message) {
    upd_mgr_.mutable_update()->ResetFeedUpdate();
    upd_mgr_.mutable_update()->SetTimestamp(message.timestamp);
    onMessage(message.native_symbol, message.data);
    if (feed_counter_) {
      feed_counter_->ConsumeOneMsg();
    }
  }

  // Trade
  void onMessage(
      const std::string& native_symbol,
      const Trade& trade) {
    auto* _trade = upd_mgr_.GetTradeMutable();
    *_trade = trade;
    upd_mgr_.PublishTrade(native_symbol, trade.has_more_trade);
  }

  // Book
  void onMessage(
      const std::string& native_symbol,
      const BookEvent& event) {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    CHECK(book);
    // only support snapshot for now
    CHECK(event.type == BookEventType::Snapshot);
    auto tx = book->NewTransaction(event.timestamp);
    tx.Clear();
    for (const auto& pq: event.asks) {
      tx.UpdateAsk(pq.price, pq.qty);
    }
    for (const auto& pq: event.bids) {
      tx.UpdateBid(pq.price, pq.qty);
    }

    upd_mgr_.PublishBook();
  }

  // Liquidation
  void onMessage(
      const std::string& native_symbol,
      const LiquidationOrder& liquidation) {
    auto* _liquidation_order = upd_mgr_.GetLiquidationOrderMutable();
    *_liquidation_order = liquidation;
    upd_mgr_.PublishLiquidationOrder(native_symbol, liquidation.has_more_liquidation_order);
  }

  // OpenInterest
  void onMessage(
      const std::string& native_symbol,
      const OpenInterest& open_interest) {
    auto* _open_interest = upd_mgr_.GetOpenInterestMutable();
    *_open_interest = open_interest;
    upd_mgr_.PublishOpenInterest(native_symbol);
  }

  // MarkPrice
  void onMessage(
      const std::string& native_symbol,
      const MarkPrice& mark_price) {
    auto* _mark_price = upd_mgr_.GetMarkPriceMutable();
    *_mark_price = mark_price;
    upd_mgr_.PublishMarkPrice(native_symbol);
  }

  // Index
  void onMessage(
      const std::string& native_symbol,
      const Index& index) {
    auto* _index = upd_mgr_.GetIndexMutable();
    *_index = index;
    upd_mgr_.PublishIndex(native_symbol);
  }

  // FundingRate
  void onMessage(
      const std::string& native_symbol,
      const FundingRate& funding_rate) {
    auto* _funding_rate = upd_mgr_.GetFundingRateMutable();
    *_funding_rate = funding_rate;
    upd_mgr_.PublishFundingRate(native_symbol);
  }

  // IvGreeks
  void onMessage(
      const std::string& native_symbol,
      const IvGreeks& iv_greeks) {
    auto* _iv_greeks = upd_mgr_.GetIvGreeksMutable();
    *_iv_greeks = iv_greeks;
    upd_mgr_.PublishIvGreeks(native_symbol);
  }

  // Nav
  void onMessage(
      const std::string& native_symbol,
      const Nav& nav) {
    auto* _nav = upd_mgr_.GetNavMutable();
    *_nav = nav;
    upd_mgr_.PublishNav(native_symbol);
  }


  // Kline
  void onMessage(
      const std::string& native_symbol,
      const KlinesEvent& event) {
    CHECK_GT(event.klines.size(), 0);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, event.klines.front().interval);
    CHECK(klines);
    // only support add for now
    CHECK(event.type == KlineEventType::Add);

    for (auto& kline : event.klines) {
      auto result = klines->AddKline(kline.open_timestamp);
      if (!result) continue;

      Kline* _kline;
      bool freezed;
      std::tie(_kline, freezed) = result.value();
      *_kline = kline;
    }
    klines->Freeze();
    upd_mgr_.PublishKline(native_symbol);
  }

  // Wallet
  void onMessage(
      const std::string& native_symbol,
      const Wallet& wallet) {
    auto* _wallet = upd_mgr_.GetWalletMutable();
    *_wallet = wallet;
    upd_mgr_.PublishWallet(native_symbol);
  }

  // MarketAnalysis
  void onMessage(
      const std::string& native_symbol,
      const MarketAnalysis& market_analysis) {
    auto* _market_analysis = upd_mgr_.GetMarketAnalysisMutable();
    *_market_analysis = market_analysis;
    upd_mgr_.PublishMarketAnalysis(native_symbol);
  }

  // TradingData
  void onMessage(
      const std::string& native_symbol,
      const TradingDataEvent<TopLongShortAccountRatio>& event) {
    auto* trading_data = upd_mgr_.GetTopLongShortAccountRatioMutable(
        native_symbol, event.interval);
    UpdateTradingData(event, trading_data);
    upd_mgr_.PublishTopLongShortAccountRatio();
  }
  void onMessage(
      const std::string& native_symbol,
      const TradingDataEvent<TopLongShortPositionRatio>& event) {
    auto* trading_data = upd_mgr_.GetTopLongShortPositionRatioMutable(
        native_symbol, event.interval);
    UpdateTradingData(event, trading_data);
    upd_mgr_.PublishTopLongShortPositionRatio();
  }
  void onMessage(
      const std::string& native_symbol,
      const TradingDataEvent<GlobalLongShortAccountRatio>& event) {
    auto* trading_data = upd_mgr_.GetGlobalLongShortAccountRatioMutable(
        native_symbol, event.interval);
    UpdateTradingData(event, trading_data);
    upd_mgr_.PublishGlobalLongShortAccountRatio();
  }
  void onMessage(
      const std::string& native_symbol,
      const TradingDataEvent<TakerBuySellRatio>& event) {
    auto* trading_data = upd_mgr_.GetTakerBuySellRatioMutable(
        native_symbol, event.interval);
    UpdateTradingData(event, trading_data);
    upd_mgr_.PublishTakerBuySellRatio();
  }

  void onMessage(
      const std::string& native_symbol,
      const TradingDataEvent<OpenInterest>& event) {
    auto* trading_data = upd_mgr_.GetOpenInterestHistMutable(
        native_symbol, event.interval);
    UpdateTradingData(event, trading_data);
    upd_mgr_.PublishOpenInterestHist();
  }


 protected:
  // trading data from event
  template<typename T>
  void UpdateTradingData(
      const TradingDataEvent<T>& event,
      TradingData<T>* trading_data) {
    CHECK_GT(event.data.size(), 0);
    // only support add for now
    CHECK(event.type == TradingDataEventType::Add);

    for (auto& item : event.data) {
      auto result = trading_data->Add(item.timestamp);
      if (!result) continue;

      T* data_ptr;
      bool freezed;
      std::tie(data_ptr, freezed) = result.value();
      *data_ptr = item;
    }
    trading_data->Freeze();
  }

  FeedUpdateManager upd_mgr_;
  IFeedSubscriber* subscriber_ = nullptr;
  FeedCounter* feed_counter_ = nullptr;
};

class ReplayTelemetryProcessor {
 public:
  ReplayTelemetryProcessor(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      FeedCounter* counter)
      : replay_processor_(
            system_name,
            mea,
            counter) {
  }

  const FeedUpdateManager& upd_mgr() const {
    return replay_processor_.upd_mgr();
  }
  FeedUpdateManager* mutable_upd_mgr() {
    return replay_processor_.mutable_upd_mgr();
  }

  virtual void AddProduct(const IProduct& product, IFeedSubscriber* subscriber) {
    replay_processor_.AddProduct(product, subscriber);
  }

  virtual void RemoveProduct(const std::vector<const IProduct*>& products) {
    replay_processor_.RemoveProduct(products);
  }

  template<typename T>
  void onSharedMessage(std::shared_ptr<ReplayLifecycleMessage<T>> message) {
    auto& _lts = replay_processor_.mutable_upd_mgr()->mutable_update()->MutableTs();
    _lts = message->lts;
    // overwrite the main, the origin main timestamp is feed parser timestamp
    int64_t ts = GetCurrentTimestamp();
    _lts.set_main_rx_timestamp(ts);
    _lts.set_main_tx_timestamp(ts);

    replay_processor_.onMessage(message->data);
  }

 protected:
  ReplayProcessor replay_processor_;
};

}  // namespace impl

using impl::ReplayProcessor;
using impl::ReplayTelemetryProcessor;
using impl::ReplayMessage;
using impl::ReplayLifecycleMessage;

}  // namespace coin2::exchange::base::feed::replay
